package com.lili.collection.mq.plugins;

import com.lili.collection.core.context.BatchRecordContext;
import com.lili.collection.core.context.StageContext;
import com.lili.collection.core.domain.RunnerMsgTransfer;
import com.lili.collection.core.enums.DataFormat;
import com.lili.collection.core.error.StageRecordException;
import com.lili.collection.core.plugins.OutputPlugin;
import com.lili.collection.core.record.Record;
import com.lili.collection.core.runner.result.ErrOutputData;
import com.lili.collection.core.runner.result.ErrorDetail;
import com.lili.collection.core.runner.result.RecordInfo;
import com.lili.collection.core.runner.result.RunnerResult;
import com.lili.collection.core.utils.HandlerUtil;
import com.lili.collection.core.utils.NoticeUtil;
import com.lili.collection.mq.config.RocketmqOutputConfig;
import com.lili.collection.mq.domain.ByteMessageWrapper;
import com.lili.collection.mq.utils.JsonConvertUtil;
import com.lili.collection.mq.utils.MQUtil;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.common.message.Message;

import java.nio.charset.StandardCharsets;
import java.util.List;

/**
 *描述: Rocketmq输出插件
 *@date: 2023/6/3  15:35
 *@author: lili
 */
@Slf4j
public class RocketmqOutputPlugin implements OutputPlugin<RocketmqOutputConfig> {
    private RocketmqOutputConfig config;
    private Throwable tempThrowable = null;
    private int executeNum;
    @Getter
    private RunnerResult runnerResult;
    @Getter
    private RunnerMsgTransfer transfer;
    private DefaultMQProducer producer;
    private int maxMessageSize;

    @Override
    public void init(RocketmqOutputConfig config) {
        try {
            this.config=config;
            if(this.config.getDataFormat()==null){
                throw new IllegalArgumentException("rocketmq输入输出数据格式定义不能为空！");
            }
            DefaultMQProducer producer = new DefaultMQProducer(config.getGroup());
            producer.setNamesrvAddr(config.getNamesrvAddr());
            producer.setRetryTimesWhenSendFailed(0);
            producer.setRetryTimesWhenSendAsyncFailed(0);
            producer.start();
            this.producer=producer;
            this.maxMessageSize = producer.getMaxMessageSize();
        } catch (MQClientException e) {
            log.error("创建rocketmq连接异常!",e);
            throw new RuntimeException(e);
        }
    }

    @Override
    public void setRunnerResult(RunnerResult outputRunnerResult) {
        this.runnerResult = outputRunnerResult;
    }

    @Override
    public void setRunnerTransfer(RunnerMsgTransfer transfer) {
        this.transfer=transfer;
    }

    @Override
    public synchronized void execute(StageContext stageContext) {
        BatchRecordContext recordContext = (BatchRecordContext) stageContext;
        RecordInfo recordInfo = new RecordInfo();
        this.executeNum++;
        boolean isCommit=false;
        RunnerResult<String> outputRunnerResult = this.getRunnerResult();
        boolean isRollback = false;
        List<Record> output = recordContext.getOutput();
        try {
            try {
                if(this.getTransfer()!=null && this.getTransfer().isRunError()){
                    throw new IllegalStateException("Rabbitmq异步运行错误!");
                }
                if (output.size() == 0) {
                    MQUtil.shutdown(this.producer);
                    log.info("-->所有数据写入完成!");
                    return;
                }
                if(config.getDataFormat()== DataFormat.ORIGIN ||  config.getDataFormat()==DataFormat.ORIGIN_CONVERT){
                    for (Record record : output) {
                        String fieldValue = String.valueOf(record.getField().getValue());
                        if(!this.sendDataToMQ(fieldValue)){
                            recordInfo.incrementErrorCount();
                            String errorMsg="Rocketmq消息发送失败!";
                            recordInfo.getErrOutputData().putErrorInfo(errorMsg);
                            recordInfo.addErrorValueDetail(new ErrorDetail("JSON",fieldValue,errorMsg,(fieldValue!=null ? fieldValue.length():0)));
                            throw new IllegalStateException("发送Rocketmq数据失败!");
                        }else {
                            recordInfo.incrementSuccessCount();
                        }
                    }
                }else {
                    List<ByteMessageWrapper> wrapperList = JsonConvertUtil.convertRecordToJsonByteWrapper(output, recordInfo, this.maxMessageSize,false);
                    for (ByteMessageWrapper byteMessageWrapper : wrapperList) {
                        //消息推送
                        if(HandlerUtil.sendProcessedData(recordInfo,config,byteMessageWrapper.getOrginJsonData(),(data)-> this.sendDataToMQ(data))){
                            recordInfo.addErrorCount(byteMessageWrapper.getDataRowSize());
                            String errorMsg="Rocketmq消息发送失败!";
                            recordInfo.addErrorValueDetail(new ErrorDetail("JSON",byteMessageWrapper.getOrginJsonData(),errorMsg,byteMessageWrapper.getBytes().length));
                            throw new IllegalStateException("发送Rocketmq数据失败!");
                        }else {
                            recordInfo.addSuccessCount(byteMessageWrapper.getDataRowSize());
                        }
                    }
                }
                //转换消息
                log.info("-->第{}批次数据写入完成\r\n-->写入量：{}",recordContext.getIndex(),output.size());
            } catch (Throwable throwable) {
                isRollback = true;
                log.error("写入rocketmq错误!",throwable);
                recordInfo.getErrOutputData().putErrorInfo(throwable.toString());
                outputRunnerResult.setSuccessed(false);
                throw throwable;
            }
        } catch (Throwable throwable) {
            tempThrowable = throwable;
            RunnerMsgTransfer transfer = this.getTransfer();
            if(transfer!=null){
                transfer.setRunError(true);
                transfer.interruptRunner();
            }
            recordContext.setStageRecordState(StageContext.StageRecordState.ALL_ERROR);
            recordContext.setException(new StageRecordException(throwable));
            outputRunnerResult.setSuccessed(false);
            //错误日志收集
            ErrOutputData errOutputData = recordInfo.getErrOutputData();
            errOutputData.setBatchIndex(recordContext.getIndex());
            errOutputData.setErrorMessage(throwable.toString());
            errOutputData.setFirstBatchError(executeNum == 1 ? true : false);
            errOutputData.setType("BATCH-OUTPUT-ROCKETMQ");
            log.info("-->第{}批次数据写入时发生异常\r\n-->原因：{}",
                    recordContext.getIndex(),
                    throwable.getLocalizedMessage(), throwable);
        }
        if (tempThrowable != null || !recordInfo.isSuccessed()) {
            outputRunnerResult.mergeErrorRecord(recordInfo.getErrOutputData());
        }
        this.addResultCount(recordInfo, outputRunnerResult);
        recordContext.clearOutput();
    }

    private boolean sendDataToMQ(String data) throws Exception{
        if(StringUtils.isBlank(data)){
            throw new IllegalArgumentException("发送MQ数据不能为空！");
        }
        return NoticeUtil.sendDataAndSaveNotice(config,data,(sendData)->{
            SendResult sendResult = producer.send(new Message(this.config.getTopic(),null, sendData.getBytes(StandardCharsets.UTF_8)));
            return sendResult.getSendStatus() == SendStatus.SEND_OK;
        });
    }

    private void addResultCount(RecordInfo recordInfo, RunnerResult<String> outputRunnerResult) {
        outputRunnerResult.addSuccessCount(recordInfo.getSuccessRecordCount());
        outputRunnerResult.addTotalCount(recordInfo.getTotalRecordCount());
        outputRunnerResult.addErrorCount(recordInfo.getErrorRecordCount());
        if(outputRunnerResult.isSuccessed()){
            outputRunnerResult.setSuccessed(recordInfo.isSuccessed());
        }
    }

    @Override
    public void dispose() {
        MQUtil.shutdown(this.producer);
    }
}